package com.tcoiss.datafactory.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tcoiss.common.core.enums.ErroEnums;
import com.tcoiss.common.core.exception.CustomException;
import com.tcoiss.common.core.utils.DateUtils;
import com.tcoiss.common.component.api.model.SchemeVO;
import com.tcoiss.datafactory.domain.BusTable;
import com.tcoiss.datafactory.domain.ExecuteWork;
import com.tcoiss.datafactory.mapper.ExecuteWorkMapper;
import com.tcoiss.datafactory.service.*;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * 作业Service业务层处理
 *
 * @author zw
 * @date 2021-04-26
 */
@Service
public class  ExecuteWorkServiceImpl extends ServiceImpl<ExecuteWorkMapper, ExecuteWork> implements IExecuteWorkService {

    @Autowired
    protected IBusTableService iBusTableService;

    @Autowired
    private IDynamicSqlService iDynamicSqlService;

    @Autowired
    private IKJobService ikJobService;

    @Autowired
    private IKTransService iKTransService;

    @Override
    public List<ExecuteWork> queryList(ExecuteWork executeWork) {
        LambdaQueryWrapper<ExecuteWork> lqw = Wrappers.lambdaQuery();
        if (executeWork.getSchemeId() != null) {
            lqw.eq(ExecuteWork::getSchemeId, executeWork.getSchemeId());
        }
        if (StringUtils.isNotBlank(executeWork.getWorkName())) {
            lqw.like(ExecuteWork::getWorkName, executeWork.getWorkName());
        }
        if (StringUtils.isNotBlank(executeWork.getTableName())) {
            lqw.like(ExecuteWork::getTableName, executeWork.getTableName());
        }
        if (executeWork.getWorkType() != null) {
            lqw.eq(ExecuteWork::getWorkType, executeWork.getWorkType());
        }
        if (executeWork.getStatus() != null) {
            lqw.eq(ExecuteWork::getStatus, executeWork.getStatus());
        }
        lqw.orderByDesc(ExecuteWork::getCreateTime);
        List<ExecuteWork> ewList = this.list(lqw);
        ewList.sort(Comparator.comparingInt(ExecuteWork::getWorkOrder));
        return ewList;
    }

    // 查询当前方案作业优先级最大值并自增
    @Override
    public Integer getWorkOderMax(Long schemeId) {
        LambdaQueryWrapper<ExecuteWork> lqw = Wrappers.lambdaQuery();
        if (schemeId != null) {
            lqw.eq(ExecuteWork::getSchemeId, schemeId);

        }
        lqw.orderByDesc(ExecuteWork::getWorkOrder);
        List<ExecuteWork> list = this.list(lqw);
        if (list.size() != 0) {
            Integer i = list.get(0).getWorkOrder();
            if (i != null) {
                return ++i;
            }
        }
        return 1;
    }

    // 获取当前作业详情
    @Override
    public ExecuteWork getByWorkId(Long workId) {
        LambdaQueryWrapper<ExecuteWork> lqw = Wrappers.lambdaQuery();
        if (workId != null) {
            lqw.eq(ExecuteWork::getWorkId, workId);
        }
        List<ExecuteWork> ewList = this.list(lqw);
        ExecuteWork ew = ewList.get(0);
        List<ExecuteWork> ews = getWorkOder(ew.getSchemeId());
        ew.setWorkOderS(getPriorityArray(ews));
        return ew;
    }

    // 获取当前方案的所有作业
    private List<ExecuteWork> getWorkOder(Long schemeId) {
        LambdaQueryWrapper<ExecuteWork> lqw = Wrappers.lambdaQuery();
        if (schemeId != null) {
            lqw.eq(ExecuteWork::getSchemeId, schemeId);
        }
        lqw.orderByDesc(ExecuteWork::getWorkOrder);
        return this.list(lqw);
    }

    // 给优先级列表赋值
    private Integer[] getPriorityArray(List<ExecuteWork> list) {
        try {
            if (list != null) {
                //给优先级列表赋值
                Integer[] workOderS = new Integer[list.size()];
                for (int i = 0; i < list.size(); i++) {
                    workOderS[i] = list.get(i).getWorkOrder();
                }
                return workOderS;
            }
        } catch (CustomException e) {
            throw new CustomException(ErroEnums.DATAERRO.getCode(),ErroEnums.DATAERRO.getInfo() + "查询优先列表失败：" + e.getLocalizedMessage(),e);
        }
        return null;
    }

    // 获取当前方案Id的所有作业优先级列表
    @Override
    public Integer[] getPriorityById(Long schemeId) {
        List<ExecuteWork> ews = getWorkOder(schemeId);
        return getPriorityArray(ews);
    }

    // 更新使用状态
    @Override
    public boolean updateStatus(List<Long> idList, Integer status) {
        if (idList != null & status != null) {
            for (Long id : idList) {
                ExecuteWork ew = getById(id);
                if (ew.getStatus() != status) {
                    ew.setStatus(status);
                    this.updateById(ew);
                }
            }
            return true;
        }
        return false;
    }

    // 根据Id修改优先级
    @Override
    public boolean updateWorkById(ExecuteWork executeWork) {
        LambdaQueryWrapper<ExecuteWork> lqw = Wrappers.lambdaQuery();
        if (executeWork.getWorkId() != null) {
            lqw.eq(ExecuteWork::getWorkId, executeWork.getWorkId());
        }
        List<ExecuteWork> work = this.list(lqw);
        return WorkSort(executeWork, work.get(0).getWorkOrder(), executeWork.getSchemeId());
    }

    // 作业重排序
    private boolean WorkSort(ExecuteWork executeWork, Integer oldWorkOrder, Long schemeId) {
        Integer workOrder = executeWork.getWorkOrder();
        List<ExecuteWork> exList = getWorkOder(schemeId);
        for (ExecuteWork ew : exList) {
            if (ew.getWorkOrder().equals(workOrder)) {
                ew.setWorkOrder(oldWorkOrder);
                this.updateById(ew);
                break;
            }
        }
        return this.updateById(executeWork);
    }

    @Override
    public List<ExecuteWork> getWorksBySchemeId(Long schemeId) {
        LambdaQueryWrapper<ExecuteWork> lqw = Wrappers.lambdaQuery();
        if (schemeId != null) {
            lqw.eq(ExecuteWork::getSchemeId, schemeId);
            lqw.eq(ExecuteWork::getStatus, 1);
        }
        lqw.orderByAsc(ExecuteWork::getWorkOrder);
        return this.list(lqw);
    }

    @Override
    public boolean executeWork(ExecuteWork work, SchemeVO vo) {
        //判断作业类型 1.数据同步 3.执行存储过程 4.执行ETL 5.数据转换
        BusTable table = iBusTableService.getBusTableByName(work.getTableName());
        String script = work.getWorkScript();
//        if(StringUtils.isBlank(script) && work.getWorkType() == 1){
//            throw new CustomException(ErroEnums.DATAERRO.getCode(), new Object[]{work.getWorkName()}, ErroEnums.DATAERRO.getInfo()+"当前作业脚本为空");
//        }
        int num = 0;
        Date lastTime = vo.getLastExecuteTime();
        try {
            switch (work.getWorkType()) {
                case 1:
                    //api 同步 存在执行日志，且作业脚本不为空是执行增量同步
                    if(lastTime!=null&&StringUtils.isNotBlank(script) ){
                        num = iBusTableService.syncAddDataByApi(table, script,lastTime);
                    }else{
                        num = iBusTableService.syncAllDataByApi(table, script);
                    }
                    break;
                case 2://调用存储过程
                    String[] params = null;
                    if(script != null && script.contains("now")){
                        script = script.replace("[", "");
                        script = script.replace("]", "");
                        script = script.replace("now", DateUtils.dateTimeNow(DateUtils.YYYY_MM_DD));
                        params = script.split(",");
                    }else{
                        if(lastTime==null){
                            params = new String[]{"1991-10-08"};
                        }else{
                            String updateTime = DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD,lastTime);
                            params = new String[]{updateTime};
                        }
                    }
                    if(table.getScriptParamNum() == 0){
                        num = iDynamicSqlService.callProcedure(table.getSqlScript(),table.getDbName(), new String[]{});
                    }else {
                        if (params.length != table.getScriptParamNum()) {
                            throw new CustomException(ErroEnums.DATAERRO.getCode(), new Object[]{work.getWorkName()}, ErroEnums.DATAERRO.getInfo()+"当前作业脚本为空");
                        }
                        num = iDynamicSqlService.callProcedure(table.getSqlScript(),table.getDbName(), params);
                        //num = iDynamicSqlService.callProcedureByParam(table.getSqlScript(), params);
                    }
                    break;
                case 3:
                    //ETL转换
                    Map<String,Object> result = iKTransService.startTransById(Long.valueOf(work.getEtlTransId()));
                    if(!result.get("code").equals("0000")){
                        throw new CustomException(ErroEnums.DATAERRO.getCode(), new Object[]{work.getWorkName(), "执行ETL转换"},
                                ErroEnums.DATAERRO.getInfo() + result.get("message"));
                    }
                    num = 1;
                    break;
                case 4:
                    // todo 关联同步
                    num = iBusTableService.syncLinkDataByApi(table, script, vo.getMsgObjectId());
                    break;
                case 5:
                    // ETL作业
                    Map<String,Object> result2 = ikJobService.startJobById(Long.valueOf(work.getEtlJobId()));
                    if(!result2.get("code").equals("0000")){
                        throw new CustomException(ErroEnums.DATAERRO.getCode(), new Object[]{work.getWorkName(), "执行ETL转换"},
                                ErroEnums.DATAERRO.getInfo() + result2.get("message"));
                    }
                    num = 1;
                    break;
                default:
                    break;
            }
        } catch (CustomException e) {
            e.printStackTrace();
            throw new CustomException(ErroEnums.DATAERRO.getCode(), new Object[]{work.getWorkName()}, "执行作业异常：" + e.getLocalizedMessage());
        } catch (Exception e) {
            e.printStackTrace();
            throw new CustomException(ErroEnums.SYSERRO.getCode(), new Object[]{work.getWorkName()}, "系统异常：" + e.getLocalizedMessage());
        }
        return num>0?true:false;

    }

    @Override
    @Async
    public void asyncExecuteWork(ExecuteWork work) {

    }
}
